package com.mtiiot.fstcpserver.es;

import com.mtiiot.fstcpserver.base.BaseService;
import com.mtiiot.fstcpserver.mq.kafka.insplatform.InstruProducer;
import com.mtiiot.fstcpserver.utils.BeanConvertUtil;
import com.mtiiot.fstcpserver.utils.Constants;
import com.mtiiot.fscore.bean.MetadataInfo;
import com.mtiiot.fscore.bean.MqInfo;
import com.mtiiot.fscore.bean.elasticsearch.metadata.BaseBean;
import com.mtiiot.fscore.inter1.ElasticsearchService1;
import com.mtiiot.fscore.utils.RequestType;
import com.mtiiot.fscore.utils.StringUtil;
import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
import org.codehaus.jackson.map.ObjectMapper;
import org.springframework.beans.factory.annotation.Autowired;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Created by lcy on 16-8-16.
 */
public class ElasticsearchServiceImpl1 extends BaseService implements ElasticsearchService1 {
    private static final Logger logger = LogManager.getLogger(ElasticsearchServiceImpl1.class);
    @Autowired
    private BaseElasticsearch baseElasticsearch;

    /**
     * 创建文档记录(同步操作)
     *
     * @param mi    元数据信息
     * @param index 文档索引名称
     * @param type  文档索引类型
     * @param bean  文档bean信息
     * @return boolean success or failure
     */
    public boolean createIndex(MetadataInfo mi, String index, String type, BaseBean bean) {
        try {
            //nosql数据入队列
            checkMetadata(mi);//验证元数据
            checkEsIndexAndType(index, type);//验证index和type
            if (StringUtil.isEmpty(type)) {
                throw new Exception("es type is empty!");
            }
            if (bean == null) {
                throw new Exception("BaseBean is empty!");
            }
            setMetadataBeanData(mi, bean, RequestType.INSERT.getName(), "");
            ObjectMapper mapper = new ObjectMapper();
            String json = mapper.writeValueAsString(bean);
            bean.setRequestSize(StringUtil.bytes2kbDouble(json.getBytes().length));
            json = mapper.writeValueAsString(bean);
            return baseElasticsearch.createIndexById(index, type, json, bean.getId());
        } catch (Exception ex) {
            //异常信息 元数据入reids队列
            handleException(Constants.MODULE_ES, "ElasticsearchServiceImpl1", "createIndex", ex, logger);
            return false;
        }
    }


    /**
     * 批量创建文档数据(同步操作)
     *
     * @param mi    元数据信息
     * @param type  对应关系型数据库的表名
     * @param beans 文档bean信息的list集合
     * @return boolean success or failure
     */
    public boolean batchCreateIndex(MetadataInfo mi, String index, String type, List<BaseBean> beans) {
        try {
            checkMetadata(mi);//验证元数据
            checkEsIndexAndType(index, type);//验证index和type
            if (StringUtil.isEmpty(type)) {
                throw new Exception("es type is empty!");
            }
            if (beans == null || beans.size() == 0) {
                throw new Exception("beans is empty!");
            }
            for (BaseBean baseBean : beans) {
                setMetadataBeanData(mi, baseBean, RequestType.INSERT.getName(), "");
                ObjectMapper mapper = new ObjectMapper();
                String json = mapper.writeValueAsString(baseBean);
                baseBean.setRequestSize(StringUtil.bytes2kbDouble(json.getBytes().length));
            }
            return baseElasticsearch.batchCreateIndexById(index, type, beans);
        } catch (Exception ex) {
            //异常信息 元数据入reids队列
            handleException(Constants.MODULE_ES, "ElasticsearchServiceImpl1", "batchCreateIndex", ex, logger);
            return false;
        }
    }

    @Override
    public boolean deleteAndCreateIndex(MetadataInfo mi, String index, String type, BaseBean bean) {
        try {
            //nosql数据入队列
            checkMetadata(mi);//验证元数据
            checkEsIndexAndType(index, type);//验证index和type
            if (StringUtil.isEmpty(type)) {
                throw new Exception("es type is empty!");
            }
            if (bean == null) {
                throw new Exception("BaseBean is empty!");
            }
            setMetadataBeanData(mi, bean, RequestType.INSERT.getName(), "");
            ObjectMapper mapper = new ObjectMapper();
            String json = mapper.writeValueAsString(bean);
            bean.setRequestSize(StringUtil.bytes2kbDouble(json.getBytes().length));
            //json=mapper.writeValueAsString(bean);
            Map<String, Object> m = BeanConvertUtil.transBean2Map(bean);
            return baseElasticsearch.updateIndex(index, type, bean.getId(), m);
            //return baseElasticsearch.updateIndexAsUpsert(index, type, bean.getId(), json);
        } catch (Exception ex) {
            //异常信息 元数据入reids队列
            handleException(Constants.MODULE_ES, "ElasticsearchServiceImpl1", "deleteAndCreateIndex", ex, logger);
            return false;
        }
    }

    @Override
    public boolean batchDeleteAndCreateIndex(MetadataInfo mi, String index, String type, List<BaseBean> beans) {
        try {
            checkMetadata(mi);//验证元数据
            checkEsIndexAndType(index, type);//验证index和type
            if (StringUtil.isEmpty(type)) {
                throw new Exception("es type is empty!");
            }
            if (beans == null || beans.size() == 0) {
                throw new Exception("beans is empty!");
            }

            Map<String, Map<String, Object>> map = new HashMap<String, Map<String, Object>>();
            for (BaseBean baseBean : beans) {
                setMetadataBeanData(mi, baseBean, RequestType.INSERT.getName(), "");
                ObjectMapper mapper = new ObjectMapper();
                String json = mapper.writeValueAsString(baseBean);
                baseBean.setRequestSize(StringUtil.bytes2kbDouble(json.getBytes().length));
                map.put(baseBean.getId(), BeanConvertUtil.transBean2Map(baseBean));
            }
            return baseElasticsearch.batchUpdateIndex(index, type, map);
            //return baseElasticsearch.batchUpdateIndexAsUpsert(index, type, beans);
        } catch (Exception ex) {
            //异常信息 元数据入reids队列
            handleException(Constants.MODULE_ES, "ElasticsearchServiceImpl1", "batchDeleteAndCreateIndex", ex, logger);
            return false;
        }
    }


    /**
     * 根据id更新数据
     *
     * @param metadataInfo
     * @param type
     * @param id
     * @param map
     * @return
     */
    @Override
    public boolean updateIndex(MetadataInfo metadataInfo, String index, String type, String id, Map<String, Object> map) {
        try {
            checkMetadata(metadataInfo);//验证元数据
            if (StringUtil.isEmpty(type)) {
                throw new Exception("es type is empty!");
            }
            if (StringUtil.isEmpty(id) || map == null) {
                throw new Exception("id or map is empty!");
            }
            return baseElasticsearch.updateIndex(index, type, id, map);
        } catch (Exception ex) {
            //异常信息 元数据入reids队列
            handleException(Constants.MODULE_ES, "ElasticsearchServiceImpl1", "updateIndex", ex, logger);
            return false;
        }
    }

    @Override
    public boolean batchUpdateIndex(MetadataInfo mi, String index, String type, Map<String, Map<String, Object>> map) {
        try {
            checkMetadata(mi);//验证元数据
            checkEsIndexAndType(index, type);//验证index和type
            if (StringUtil.isEmpty(type)) {
                throw new Exception("es type is empty!");
            }
            if (map == null || map.size() == 0) {
                throw new Exception("beans is empty!");
            }
            return baseElasticsearch.batchUpdateIndex(index, type, map);
        } catch (Exception ex) {
            //异常信息 元数据入reids队列
            handleException(Constants.MODULE_ES, "ElasticsearchServiceImpl1", "batchUpdateIndex", ex, logger);
            return false;
        }
    }


    /**
     * 根据id删除数据
     *
     * @param metadataInfo
     * @param type
     * @param id
     * @return
     */
    @Override
    public boolean deleteIndex(MetadataInfo metadataInfo, String index, String type, String id) {
        try {
            checkMetadata(metadataInfo);//验证元数据
            if (StringUtil.isEmpty(type)) {
                throw new Exception("es type is empty!");
            }
            if (StringUtil.isEmpty(id)) {
                throw new Exception("id is empty!");
            }
            return baseElasticsearch.deleteIndex(index, type, id);
        } catch (Exception ex) {
            //异常信息 元数据入reids队列
            handleException(Constants.MODULE_ES, "ElasticsearchServiceImpl1", "deleteIndex", ex, logger);
            return false;
        }
    }

    @Override
    public boolean batchDeleteIndex(MetadataInfo mi, String index, String type, List<String> ids) {
        try {
            checkMetadata(mi);//验证元数据
            checkEsIndexAndType(index, type);//验证index和type
            if (StringUtil.isEmpty(type)) {
                throw new Exception("es type is empty!");
            }
            if (ids == null || ids.size() == 0) {
                throw new Exception("beans is empty!");
            }
            return baseElasticsearch.batchDeleteIndexs(index, type, ids);
        } catch (Exception ex) {
            //异常信息 元数据入reids队列
            handleException(Constants.MODULE_ES, "ElasticsearchServiceImpl1", "batchDeleteIndex", ex, logger);
            return false;
        }
    }

    /////////////////////////////////////////////////////////////////////////////////////////////////////
    // 下面方法是异步操作
    /////////////////////////////////////////////////////////////////////////////////////////////////////

    /**
     * 创建文档记录(异步操作)
     *
     * @param metadataInfo 元数据信息
     * @param index        文档索引名称
     * @param type         文档索引类型
     * @param bean         文档bean信息
     * @return boolean success or failure
     */
    @Override
    public boolean createIndexByAsy(MetadataInfo metadataInfo, String index, String type, BaseBean bean) {
        try {
            //nosql数据入队列
            checkMetadata(metadataInfo);//验证元数据
            checkEsIndexAndType(index, type);//验证index和type
            if (StringUtil.isEmpty(type)) {
                throw new Exception("es type is empty!");
            }
            if (bean == null) {
                throw new Exception("BaseBean is empty!");
            }
            setMetadataBeanData(metadataInfo, bean, RequestType.INSERT.getName(), "");
            ObjectMapper mapper = new ObjectMapper();
            String json = mapper.writeValueAsString(bean);
            bean.setRequestSize(StringUtil.bytes2kbDouble(json.getBytes().length));
            json = mapper.writeValueAsString(bean);

            // 消息发往kafka
            String code = Constants.ES_DATA_TOPIC;
            String time = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date());
            String handle = RequestType.INSERT.getName();
            // key的格式:time$$_$$index$$_$$type$$_$$insert
            String key = time + Constants.ES_DATA_SEPARATOR + index + Constants.ES_DATA_SEPARATOR + type + Constants.ES_DATA_SEPARATOR + handle;
            InstruProducer.sendData(code, key, json);
            return true;
        } catch (Exception ex) {
            //异常信息 元数据入reids队列
            handleException(Constants.MODULE_ES, "ElasticsearchServiceImpl1", "createIndexByAsy", ex, logger);
            return false;
        }
    }

    /**
     * 批量创建文档数据(异步操作)
     *
     * @param metadataInfo 元数据信息
     * @param type         对应关系型数据库的表名
     * @param beans        文档bean信息的list集合
     * @return boolean success or failure
     */
    @Override
    public boolean batchCreateIndexByAsy(MetadataInfo metadataInfo, String index, String type, List<BaseBean> beans) {
        try {
            checkMetadata(metadataInfo);//验证元数据
            checkEsIndexAndType(index, type);//验证index和type
            if (StringUtil.isEmpty(type)) {
                throw new Exception("es type is empty!");
            }
            if (beans == null || beans.size() == 0) {
                throw new Exception("beans is empty!");
            }

            // 消息批量发送到Kafka
            Map<String, String> msgMap = new HashMap<String, String>();
            String code = Constants.ES_DATA_TOPIC;
            String time = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date());
            String handle = RequestType.INSERT.getName();
            // key的格式:time$$_$$index$$_$$type$$_$$insert
            String key = time + Constants.ES_DATA_SEPARATOR + index + Constants.ES_DATA_SEPARATOR + type + Constants.ES_DATA_SEPARATOR + handle;

            for (BaseBean baseBean : beans) {
                setMetadataBeanData(metadataInfo, baseBean, RequestType.INSERT.getName(), "");
                ObjectMapper mapper = new ObjectMapper();
                String json = mapper.writeValueAsString(baseBean);
                baseBean.setRequestSize(StringUtil.bytes2kbDouble(json.getBytes().length));
                json = mapper.writeValueAsString(baseBean);
                msgMap.put(key, json);
            }
            // 调用kafka 接口
            InstruProducer.sendDatas(code, msgMap);
            return true;
        } catch (Exception ex) {
            //异常信息 元数据入reids队列
            handleException(Constants.MODULE_ES, "ElasticsearchServiceImpl1", "batchCreateIndexByAsy", ex, logger);
            return false;
        }
    }

    /**
     * 根据id更新数据(异步操作)
     *
     * @param metadataInfo
     * @param type
     * @param id
     * @param map
     * @return
     */
    @Override
    public boolean updateIndexByAsy(MetadataInfo metadataInfo, String index, String type, String id, Map<String, Object> map) {
        try {
            checkMetadata(metadataInfo);//验证元数据
            if (StringUtil.isEmpty(type)) {
                throw new Exception("es type is empty!");
            }
            if (StringUtil.isEmpty(id) || map == null) {
                throw new Exception("id or map is empty!");
            }

            // 消息批量发送到Kafka
            String code = Constants.ES_DATA_TOPIC;
            String time = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date());
            String handle = RequestType.UPDATE.getName();
            // key的格式:time$$_$$index$$_$$type$$_$$insert
            // 注意: key 后加多一个更新的ID
            String key = time + Constants.ES_DATA_SEPARATOR + index + Constants.ES_DATA_SEPARATOR + type + Constants.ES_DATA_SEPARATOR + handle + Constants.ES_DATA_SEPARATOR + id;
            ObjectMapper mapper = new ObjectMapper();
            String msg = mapper.writeValueAsString(map);
            InstruProducer.sendData(code, key, msg);
            return true;
        } catch (Exception ex) {
            //异常信息 元数据入reids队列
            handleException(Constants.MODULE_ES, "ElasticsearchServiceImpl1", "updateIndexByAsy", ex, logger);
            return false;
        }
    }

    /**
     * 根据id批量更新数据(异步操作)
     *
     * @param metadataInfo
     * @param type
     * @param map
     * @return
     */
    @Override
    public boolean batchUpdateIndexByAsy(MetadataInfo metadataInfo, String index, String type, Map<String, Map<String, Object>> map) {
        try {
            checkMetadata(metadataInfo);//验证元数据
            checkEsIndexAndType(index, type);//验证index和type
            if (StringUtil.isEmpty(type)) {
                throw new Exception("es type is empty!");
            }
            if (map == null || map.size() == 0) {
                throw new Exception("beans is empty!");
            }

            // 消息批量发送到Kafka
            Map<String, String> msgMap = new HashMap<String, String>();
            String code = Constants.ES_DATA_TOPIC;
            String time = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date());
            String handle = RequestType.UPDATE.getName();
            // key的格式:time$$_$$index$$_$$type$$_$$insert
            //String key =time+"$$_$$"+index+"$$_$$"+type+"$$_$$"+handle;
            ObjectMapper mapper = new ObjectMapper();
            for (Map.Entry<String, Map<String, Object>> m : map.entrySet()) {
                String id = m.getKey();
                String key = time + Constants.ES_DATA_SEPARATOR + index + Constants.ES_DATA_SEPARATOR + type + Constants.ES_DATA_SEPARATOR + handle + Constants.ES_DATA_SEPARATOR + id;
                Map<String, Object> objectMap = new HashMap<String, Object>();
                String json = mapper.writeValueAsString(objectMap);
                msgMap.put(key, json);
            }
            InstruProducer.sendDatas(code, msgMap);
            return true;
        } catch (Exception ex) {
            //异常信息 元数据入reids队列
            handleException(Constants.MODULE_ES, "ElasticsearchServiceImpl1", "batchUpdateIndexByAsy", ex, logger);
            return false;
        }
    }

    /**
     * 根据id删除数据(异步操作)
     *
     * @param metadataInfo
     * @param type
     * @return
     */
    @Override
    public boolean deleteIndexByAsy(MetadataInfo metadataInfo, String index, String type, String id) {
        try {
            checkMetadata(metadataInfo);//验证元数据
            if (StringUtil.isEmpty(type)) {
                throw new Exception("es type is empty!");
            }
            if (StringUtil.isEmpty(id)) {
                throw new Exception("id is empty!");
            }
            // 消息批量发送到Kafka
            String code = Constants.ES_DATA_TOPIC;
            String time = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date());
            String handle = RequestType.DELETE.getName();
            // key的格式:time$$_$$index$$_$$type$$_$$insert
            // 注意: key 后加多一个更新的ID
            String key = time + Constants.ES_DATA_SEPARATOR + index + Constants.ES_DATA_SEPARATOR + type + Constants.ES_DATA_SEPARATOR + handle;
            InstruProducer.sendData(code, key, id);
            return true;
        } catch (Exception ex) {
            //异常信息 元数据入reids队列
            handleException(Constants.MODULE_ES, "ElasticsearchServiceImpl1", "deleteIndex", ex, logger);
            return false;
        }
    }

    /**
     * 根据id批量删除数据 (异步操作)
     *
     * @param metadataInfo
     * @param type
     * @return
     */
    @Override
    public boolean batchDeleteIndexByAsy(MetadataInfo metadataInfo, String index, String type, List<String> ids) {
        try {
            checkMetadata(metadataInfo);//验证元数据
            checkEsIndexAndType(index, type);//验证index和type
            if (StringUtil.isEmpty(type)) {
                throw new Exception("es type is empty!");
            }
            if (ids == null || ids.size() == 0) {
                throw new Exception("beans is empty!");
            }
            // 消息批量发送到Kafka
            Map<String, String> msgMap = new HashMap<String, String>();
            String code = Constants.ES_DATA_TOPIC;
            String time = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date());
            String handle = RequestType.DELETE.getName();
            // key的格式:time$$_$$index$$_$$type$$_$$insert
            String key = time + Constants.ES_DATA_SEPARATOR + index + Constants.ES_DATA_SEPARATOR + type + Constants.ES_DATA_SEPARATOR + handle;
            for (String id : ids) {
                msgMap.put(key, id);
            }
            InstruProducer.sendDatas(code, msgMap);
            return true;
        } catch (Exception ex) {
            //异常信息 元数据入reids队列
            handleException(Constants.MODULE_ES, "ElasticsearchServiceImpl1", "batchDeleteIndexByAsy", ex, logger);
            return false;
        }
    }
}
